package io.github.wanggit.antrpc.client.zk.listener;

import io.github.wanggit.antrpc.client.zk.IZkClient;
import io.github.wanggit.antrpc.client.zk.register.IZkRegisterHolder;
import io.github.wanggit.antrpc.commons.constants.ConstantValues;
import io.github.wanggit.antrpc.commons.utils.JsonUtils;
import io.github.wanggit.antrpc.commons.zookeeper.NodeUnregDTO;
import io.github.wanggit.antrpc.commons.zookeeper.ZkNodeType;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;

@Slf4j
public class UnregNodeChangeListener implements IListener {

    private final IZkClient zkClient;
    private final IZkRegisterHolder zkRegisterHolder;

    public UnregNodeChangeListener(IZkClient zkClient, IZkRegisterHolder zkRegisterHolder) {
        this.zkClient = zkClient;
        this.zkRegisterHolder = zkRegisterHolder;
    }

    @Override
    public void listen() {
        internalListen(zkClient.getCurator());
    }

    /**
     *
     *
     * <pre>
     *     <[CDATA[
     *     /__rpc_unreg__/192.168.1.20:6060/io.github.wanggit.antrpc.demo.api.HelloService
     *     {
     *         ts: 1223452342423,
     *         host: "192.168.1.20:6060",
     *         clazz: "io.github.wanggit.antrpc.demo.api.HelloService"
     *     }
     *     ]]>
     * </pre>
     *
     * @param curatorFramework curatorFramework
     */
    private void internalListen(CuratorFramework curatorFramework) {
        TreeCache treeCache =
                new TreeCache(curatorFramework, "/" + ConstantValues.ZK_ROOT_NODE_UNREG_NAME);
        treeCache
                .getListenable()
                .addListener(
                        (client, event) -> {
                            ChildData childData = event.getData();
                            if (null == childData) {
                                return;
                            }
                            String path = childData.getPath();
                            ZkNodeType.UnregNodeType unregNodeType =
                                    ZkNodeType.getUnregNodeType(path);
                            if (ZkNodeType.UnregNodeType.INTERFACE.equals(unregNodeType)) {
                                byte[] data = childData.getData();
                                NodeUnregDTO nodeUnregDTO =
                                        JsonUtils.fromJson(data, NodeUnregDTO.class);
                                if (TreeCacheEvent.Type.NODE_ADDED.equals(event.getType())
                                        || TreeCacheEvent.Type.NODE_UPDATED.equals(
                                                event.getType())) {
                                    zkRegisterHolder.setRegisterBeanPausedValue(
                                            nodeUnregDTO.getClazz(), nodeUnregDTO.getHost(), true);
                                } else if (TreeCacheEvent.Type.NODE_REMOVED.equals(
                                        event.getType())) {
                                    zkRegisterHolder.setRegisterBeanPausedValue(
                                            nodeUnregDTO.getClazz(), nodeUnregDTO.getHost(), false);
                                }
                            }
                        });
        try {
            treeCache.start();
        } catch (Exception e) {
            String errorMessage = "TreeCache failed to start.";
            if (log.isErrorEnabled()) {
                log.error(errorMessage, e);
            }
            throw new ZkListenerException(errorMessage, e);
        }
    }
}
